#include <assert.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/ip.h>
#include <time.h>
#include <math.h>
#include <vector>
#include <string>
// proj
#include "common.h"
#include "hashtable.h"
#include "zset.h"
#include "heap.h"
#include "thread_pool.h"
#include "list.h"

/**
 * @brief 常量定义
 */
static const size_t k_max_msg = 4096;
static const size_t k_max_args = 1024;

/**
 * @brief 连接状态枚举
 */
enum {
    STATE_REQ = 0,  // 读取请求
    STATE_RES = 1,  // 发送响应
    STATE_END = 2,  // 标记删除
};

/**
 * @brief 连接结构体
 */
struct Conn {
    int fd = -1;
    uint32_t state = 0;     // STATE_REQ 或 STATE_RES
    uint32_t conn_id = 0;   // 连接ID
    // 读缓冲区
    size_t rbuf_size = 0;
    uint8_t rbuf[4 + k_max_msg];
    // 写缓冲区
    size_t wbuf_size = 0;
    size_t wbuf_sent = 0;
    uint8_t wbuf[4 + k_max_msg];
    
    uint64_t idle_start = 0;
    // 计时器
    DList idle_list;
};

/**
 * @brief 全局数据
 */
static struct {
    HMap db;
    // 计时器
    std::vector<HeapItem> heap;
    // 空闲计时器
    DList idle_list;
    // 线程池
    ThreadPool tp;
    // 连接计数器
    uint32_t next_conn_id = 1;
} g_data;

/**
 * @brief 数据库条目类型
 */
enum {
    T_STR = 0,
    T_ZSET = 1,
};

/**
 * @brief 数据库条目结构
 */
struct Entry {
    struct HNode node;
    std::string key;
    std::string val;
    uint32_t type = 0;
    ZSet *zset = NULL;
    // TTL相关
    size_t heap_idx = -1;
    uint64_t expire_at = 0;
};

/**
 * @brief 错误码
 */
enum {
    ERR_UNKNOWN = 1,
    ERR_2BIG = 2,
    ERR_TYPE = 3,
    ERR_ARG = 4,
};

/**
 * @brief 序列化类型
 */
enum {
    SER_NIL = 0,    // 空值
    SER_ERR = 1,    // 错误码 + 消息
    SER_STR = 2,    // 字符串
    SER_INT = 3,    // int64
    SER_DBL = 4,    // double
    SER_ARR = 5,    // 数组
};

// 前向声明
static void die(const char *msg);
static void msg(const char *msg);
static bool str2dbl(const std::string &s, double &out);
static bool str2int(const std::string &s, int64_t &out);
static bool cmd_is(const std::string &word, const char *cmd);
static void state_res(Conn *conn);

/**
 * @brief 设置文件描述符为非阻塞模式
 * @param fd 文件描述符
 */
static void fd_set_nb(int fd) {
    errno = 0;
    int flags = fcntl(fd, F_GETFL, 0);
    if (errno) {
        die("fcntl error");
        return;
    }

    flags |= O_NONBLOCK;

    errno = 0;
    (void)fcntl(fd, F_SETFL, flags);
    if (errno) {
        die("fcntl error");
    }
}

/**
 * @brief 错误处理函数
 * @param msg 错误消息
 */
static void die(const char *msg) {
    int err = errno;
    fprintf(stderr, "[%d] %s\n", err, msg);
    abort();
}

/**
 * @brief 消息处理函数
 * @param msg 消息
 */
static void msg(const char *msg) {
    fprintf(stderr, "%s\n", msg);
}

/**
 * @brief 获取当前时间戳（毫秒）
 * @return 时间戳
 */
static uint64_t get_monotonic_usec() {
    timespec tv = {0, 0};
    clock_gettime(CLOCK_MONOTONIC, &tv);
    return uint64_t(tv.tv_sec) * 1000000 + tv.tv_nsec / 1000;
}

/**
 * @brief 从缓冲区读取32位整数
 * @param buf 缓冲区
 * @return 32位整数
 */
static uint32_t read_le32(const uint8_t *buf) {
    uint32_t val = 0;
    memcpy(&val, buf, 4);
    return val;
}

/**
 * @brief 向缓冲区写入32位整数
 * @param buf 缓冲区
 * @param val 32位整数
 */
static void write_le32(uint8_t *buf, uint32_t val) {
    memcpy(buf, &val, 4);
}

/**
 * @brief 向缓冲区写入64位整数
 * @param buf 缓冲区
 * @param val 64位整数
 */
static void write_le64(uint8_t *buf, uint64_t val) {
    memcpy(buf, &val, 8);
}

/**
 * @brief 向缓冲区写入双精度浮点数
 * @param buf 缓冲区
 * @param val 双精度浮点数
 */
static void write_dbl(uint8_t *buf, double val) {
    memcpy(buf, &val, 8);
}

/**
 * @brief 从缓冲区读取64位整数
 * @param buf 缓冲区
 * @return 64位整数
 */
static uint64_t read_le64(const uint8_t *buf) {
    uint64_t val = 0;
    memcpy(&val, buf, 8);
    return val;
}

/**
 * @brief 从缓冲区读取双精度浮点数
 * @param buf 缓冲区
 * @return 双精度浮点数
 */
static double read_dbl(const uint8_t *buf) {
    double val = 0;
    memcpy(&val, buf, 8);
    return val;
}

/**
 * @brief 输出空值
 * @param out 输出缓冲区
 */
static void out_nil(std::string &out) {
    out.push_back(SER_NIL);
}

/**
 * @brief 输出字符串
 * @param out 输出缓冲区
 * @param s 字符串
 */
static void out_str(std::string &out, const std::string &s) {
    out.push_back(SER_STR);
    uint32_t len = (uint32_t)s.size();
    out.append((char *)&len, 4);
    out.append(s);
}

/**
 * @brief 输出整数
 * @param out 输出缓冲区
 * @param val 整数值
 */
static void out_int(std::string &out, int64_t val) {
    out.push_back(SER_INT);
    out.append((char *)&val, 8);
}

/**
 * @brief 输出双精度浮点数
 * @param out 输出缓冲区
 * @param val 双精度浮点数值
 */
static void out_dbl(std::string &out, double val) {
    out.push_back(SER_DBL);
    out.append((char *)&val, 8);
}

/**
 * @brief 输出错误
 * @param out 输出缓冲区
 * @param code 错误码
 * @param msg 错误消息
 */
static void out_err(std::string &out, int32_t code, const std::string &msg) {
    out.push_back(SER_ERR);
    out.append((char *)&code, 4);
    uint32_t len = (uint32_t)msg.size();
    out.append((char *)&len, 4);
    out.append(msg);
}

/**
 * @brief 开始输出数组
 * @param out 输出缓冲区
 * @param n 数组元素个数
 */
static void out_arr(std::string &out, uint32_t n) {
    out.push_back(SER_ARR);
    out.append((char *)&n, 4);
}

/**
 * @brief 更新TTL堆
 * @param entry 数据库条目
 * @param expire_at 过期时间
 */
static void entry_set_ttl(Entry *entry, uint64_t expire_at) {
    if (expire_at < UINT64_MAX && entry->heap_idx == (size_t)-1) {
        // 添加到堆中
        HeapItem item;
        item.ref = &entry->heap_idx;
        item.val = expire_at;
        g_data.heap.push_back(item);
        entry->heap_idx = g_data.heap.size() - 1;
        heap_update(g_data.heap.data(), entry->heap_idx, g_data.heap.size());
    } else if (expire_at < UINT64_MAX) {
        // 更新堆中的值
        g_data.heap[entry->heap_idx].val = expire_at;
        heap_update(g_data.heap.data(), entry->heap_idx, g_data.heap.size());
    } else if (entry->heap_idx != (size_t)-1) {
        // 从堆中移除
        size_t pos = entry->heap_idx;
        g_data.heap[pos] = g_data.heap.back();
        *g_data.heap[pos].ref = pos;
        g_data.heap.pop_back();
        if (pos < g_data.heap.size()) {
            heap_update(g_data.heap.data(), pos, g_data.heap.size());
        }
        entry->heap_idx = -1;
    }
    entry->expire_at = expire_at;
}

/**
 * @brief 删除数据库条目
 * @param entry 要删除的条目
 */
static void entry_del(Entry *entry) {
    switch (entry->type) {
    case T_ZSET:
        zset_clear(entry->zset);
        delete entry->zset;
        break;
    }
    delete entry;
}

/**
 * @brief 异步删除数据库条目
 * @param ptr 条目指针
 */
static void entry_del_async(void *ptr) {
    entry_del((Entry *)ptr);
}

/**
 * @brief 哈希表比较函数
 * @param node 哈希表节点
 * @param key 查找键
 * @return 如果相等返回true
 */
static bool entry_eq(HNode *node, HNode *key) {
    struct Entry *ent = container_of(node, struct Entry, node);
    struct Entry *k = container_of(key, struct Entry, node);
    return ent->key == k->key;
}

/**
 * @brief 检查条目是否已过期
 */
static bool entry_is_expired(Entry *ent) {
    if (ent->expire_at == 0) {
        return false;  // 没有设置过期时间
    }
    return get_monotonic_usec() >= ent->expire_at;
}

/**
 * @brief 查找数据库条目
 * @param key 键
 * @return 找到的条目，如果不存在返回NULL
 */
static Entry *entry_lookup(const std::string &key) {
    Entry k;
    k.key = key;
    k.node.hcode = str_hash((uint8_t *)key.data(), key.size());
    HNode *node = hm_lookup(&g_data.db, (uint8_t *)key.data(), key.size());
    return node ? container_of(node, Entry, node) : NULL;
}

/**
 * @brief 获取或创建数据库条目
 * @param key 键
 * @return 条目指针
 */
static Entry *entry_get(const std::string &key) {
    Entry *entry = entry_lookup(key);
    if (entry) {
        return entry;
    }

    // 创建新条目
    entry = new Entry();
    entry->key = key;
    entry->node.hcode = str_hash((uint8_t *)key.data(), key.size());
    hm_insert(&g_data.db, &entry->node);
    return entry;
}

/**
 * @brief 删除数据库条目
 * @param key 键
 * @return 如果删除成功返回true
 */
static bool entry_del_key(const std::string &key) {
    Entry k;
    k.key = key;
    k.node.hcode = str_hash((uint8_t *)key.data(), key.size());
    HNode *node = hm_delete(&g_data.db, &k.node, &entry_eq);
    if (!node) {
        return false;
    }

    Entry *entry = container_of(node, Entry, node);
    entry_set_ttl(entry, UINT64_MAX);
    
    const size_t k_large_container_size = 10000;
    bool too_big = false;
    switch (entry->type) {
    case T_ZSET:
        too_big = hm_size(&entry->zset->hmap) > k_large_container_size;
        break;
    }

    if (too_big) {
        thread_pool_queue(&g_data.tp, &entry_del_async, entry);
    } else {
        entry_del(entry);
    }
    return true;
}

/**
 * @brief 处理过期的键
 */
static void process_timers() {
    uint64_t now_us = get_monotonic_usec();
    while (!g_data.heap.empty() && g_data.heap[0].val < now_us) {
        Entry *entry = container_of(g_data.heap[0].ref, Entry, heap_idx);
        entry_del_key(entry->key);
    }
}

/**
 * @brief GET命令处理
 */
static void do_get(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (!ent) {
        return out_nil(out);
    }
    if (ent->type != T_STR) {
        return out_err(out, ERR_TYPE, "expect string type");
    }
    if (entry_is_expired(ent)) {
        return out_nil(out);
    }
    return out_str(out, ent->val);
}

/**
 * @brief SET命令处理
 */
static void do_set(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (ent) {
        if (ent->type != T_STR) {
            return out_err(out, ERR_TYPE, "expect string type");
        }
        ent->val = cmd[2];
    } else {
        ent = new Entry();
        ent->key = cmd[1];
        ent->node.hcode = str_hash((uint8_t *)ent->key.data(), ent->key.size());
        ent->val = cmd[2];
        hm_insert(&g_data.db, &ent->node);
    }
    return out_nil(out);
}

/**
 * @brief DEL命令处理
 */
static void do_del(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (ent) {
        HNode *node = hm_pop(&g_data.db, (uint8_t *)ent->key.data(), ent->key.size());
        assert(node == &ent->node);
        entry_del(ent);
    }
    return out_int(out, ent ? 1 : 0);
}

/**
 * @brief KEYS命令处理
 */
static void do_keys(std::vector<std::string> &cmd, std::string &out) {
    (void)cmd;
    out_arr(out, (uint32_t)hm_size(&g_data.db));
    
    HNode *node = NULL;
    HNode *pos = NULL;  // 改为HNode*类型
    while ((node = hm_scan(&g_data.db, &pos))) 
    {
        Entry *ent = container_of(node, Entry, node);
        if (!entry_is_expired(ent)) 
        {
            out_str(out, ent->key);
        }
    }
}

/**
 * @brief ZADD命令处理
 */
static void do_zadd(std::vector<std::string> &cmd, std::string &out) {
    double score = 0;
    if (!str2dbl(cmd[2], score)) {
        return out_err(out, ERR_ARG, "expect fp number");
    }

    // 查找或创建zset
    Entry *ent = entry_lookup(cmd[1]);
    if (ent) {
        if (ent->type != T_ZSET) {
            return out_err(out, ERR_TYPE, "expect zset");
        }
    } else {
        ent = new Entry();
        ent->key = cmd[1];
        ent->node.hcode = str_hash((uint8_t *)ent->key.data(), ent->key.size());
        ent->type = T_ZSET;
        ent->zset = new ZSet();
        zset_new(ent->zset);
        hm_insert(&g_data.db, &ent->node);
    }

    // 添加到zset
    const std::string &name = cmd[3];
    bool added = zset_add(ent->zset, name.data(), name.size(), score);
    return out_int(out, (int64_t)added);
}

/**
 * @brief ZREM命令处理
 */
static void do_zrem(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (ent && ent->type != T_ZSET) {
        return out_err(out, ERR_TYPE, "expect zset");
    }
    if (!ent) {
        return out_int(out, 0);
    }

    const std::string &name = cmd[2];
    ZNode *znode = zset_pop(ent->zset, name.data(), name.size());
    if (znode) {
        znode_del(znode);
    }
    return out_int(out, znode ? 1 : 0);
}

/**
 * @brief ZSCORE命令处理
 */
static void do_zscore(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (ent && ent->type != T_ZSET) {
        return out_err(out, ERR_TYPE, "expect zset");
    }
    if (!ent) {
        return out_nil(out);
    }

    const std::string &name = cmd[2];
    ZNode *znode = zset_lookup(ent->zset, name.data(), name.size());
    return znode ? out_dbl(out, znode->score) : out_nil(out);
}

/**
 * @brief ZQUERY命令处理
 */
static void do_zquery(std::vector<std::string> &cmd, std::string &out) {
    // zquery zset score name offset limit
    double score = 0;
    if (!str2dbl(cmd[2], score)) {
        return out_err(out, ERR_ARG, "expect fp number");
    }
    const std::string &name = cmd[3];
    int64_t offset = 0;
    int64_t limit = 0;
    if (!str2int(cmd[4], offset)) {
        return out_err(out, ERR_ARG, "expect int");
    }
    if (!str2int(cmd[5], limit)) {
        return out_err(out, ERR_ARG, "expect int");
    }

    // 获取zset
    Entry *ent = entry_lookup(cmd[1]);
    if (ent && ent->type != T_ZSET) {
        return out_err(out, ERR_TYPE, "expect zset");
    }
    if (!ent) {
        return out_arr(out, 0);
    }

    // 查询
    if (limit <= 0) {
        return out_arr(out, 0);
    }
    ZNode *znode = zset_query(ent->zset, score, name.data(), name.size());
    znode = znode_offset(znode, offset);

    // 输出
    size_t arr_pos = out.size();
    out_arr(out, 0);
    uint32_t n = 0;
    while (znode && (int64_t)n < limit) {
        out_str(out, std::string(znode->name, znode->len));
        out_dbl(out, znode->score);
        znode = znode_offset(znode, +1);
        n += 2;
    }
    // 更新数组长度
    memcpy(&out[arr_pos + 1], &n, 4);
    return;
}

/**
 * @brief PEXPIRE命令处理
 */
static void do_expire(std::vector<std::string> &cmd, std::string &out) {
    int64_t ttl_ms = 0;
    if (!str2int(cmd[2], ttl_ms)) {
        return out_err(out, ERR_ARG, "expect int64");
    }

    Entry *ent = entry_lookup(cmd[1]);
    if (ent) {
        entry_set_ttl(ent, ttl_ms);
    }
    return out_int(out, ent ? 1 : 0);
}

/**
 * @brief PTTL命令处理
 */
static void do_ttl(std::vector<std::string> &cmd, std::string &out) {
    Entry *ent = entry_lookup(cmd[1]);
    if (!ent) {
        return out_int(out, -2);
    }
    if (ent->heap_idx == (size_t)-1) {
        return out_int(out, -1);
    }

    uint64_t expire_at = g_data.heap[ent->heap_idx].val;
    uint64_t now_us = get_monotonic_usec();
    return out_int(out, expire_at > now_us ? (expire_at - now_us) / 1000 : 0);
}

/**
 * @brief 字符串转双精度浮点数
 * @param s 字符串
 * @param out 输出值
 * @return 转换成功返回true
 */
static bool str2dbl(const std::string &s, double &out) {
    char *endp = NULL;
    out = strtod(s.c_str(), &endp);
    return endp == s.c_str() + s.size() && !isnan(out) && !isinf(out);
}

/**
 * @brief 字符串转整数
 * @param s 字符串
 * @param out 输出值
 * @return 转换成功返回true
 */
static bool str2int(const std::string &s, int64_t &out) {
    char *endp = NULL;
    out = strtoll(s.c_str(), &endp, 10);
    return endp == s.c_str() + s.size();
}

/**
 * @brief 命令分发
 * @param cmd 命令参数
 * @param out 输出缓冲区
 */
static void do_request(std::vector<std::string> &cmd, std::string &out) {
    if (cmd.size() == 1 && cmd_is(cmd[0], "keys")) {
        do_keys(cmd, out);
    } else if (cmd.size() == 2 && cmd_is(cmd[0], "get")) {
        do_get(cmd, out);
    } else if (cmd.size() == 3 && cmd_is(cmd[0], "set")) {
        do_set(cmd, out);
    } else if (cmd.size() == 2 && cmd_is(cmd[0], "del")) {
        do_del(cmd, out);
    } else if (cmd.size() == 4 && cmd_is(cmd[0], "zadd")) {
        do_zadd(cmd, out);
    } else if (cmd.size() == 3 && cmd_is(cmd[0], "zrem")) {
        do_zrem(cmd, out);
    } else if (cmd.size() == 3 && cmd_is(cmd[0], "zscore")) {
        do_zscore(cmd, out);
    } else if (cmd.size() == 6 && cmd_is(cmd[0], "zquery")) {
        do_zquery(cmd, out);
    } else if (cmd.size() == 3 && cmd_is(cmd[0], "pexpire")) {
        do_expire(cmd, out);
    } else if (cmd.size() == 2 && cmd_is(cmd[0], "pttl")) {
        do_ttl(cmd, out);
    } else {
        // 未知命令
        out_err(out, ERR_UNKNOWN, "Unknown cmd");
    }
}

/**
 * @brief 命令比较（忽略大小写）
 * @param word 输入单词
 * @param cmd 命令
 * @return 如果匹配返回true
 */
static bool cmd_is(const std::string &word, const char *cmd) {
    return 0 == strcasecmp(word.c_str(), cmd);
}

/**
 * @brief 解析请求
 * @param data 请求数据
 * @param len 数据长度
 * @param cmd 解析出的命令
 * @return 解析成功返回0
 */
static int32_t parse_req(const uint8_t *data, size_t len, std::vector<std::string> &cmd) {
    if (len < 4) {
        return -1;
    }
    uint32_t n = 0;
    memcpy(&n, &data[0], 4);
    if (n > k_max_args) {
        return -1;
    }

    size_t pos = 4;
    while (n--) {
        if (pos + 4 > len) {
            return -1;
        }
        uint32_t sz = 0;
        memcpy(&sz, &data[pos], 4);
        if (pos + 4 + sz > len) {
            return -1;
        }
        cmd.push_back(std::string((char *)&data[pos + 4], sz));
        pos += 4 + sz;
    }

    if (pos != len) {
        return -1;  // 尾部垃圾
    }
    return 0;
}

/**
 * @brief 尝试一次请求
 */
static bool try_one_request(Conn *conn) {
    // 尝试解析请求
    if (conn->rbuf_size < 4) {
        // 不够4字节，无法读取长度
        return false;
    }
    uint32_t len = 0;
    memcpy(&len, &conn->rbuf[0], 4);
    if (len > k_max_msg) {
        msg("too long");
        conn->state = STATE_END;
        return false;
    }
    if (4 + len > conn->rbuf_size) {
        // 不够完整消息
        return false;
    }

    // 解析请求
    std::vector<std::string> cmd;
    if (0 != parse_req(&conn->rbuf[4], len, cmd)) {
        msg("bad req");
        conn->state = STATE_END;
        return false;
    }

    // 处理请求
    std::string out;
    do_request(cmd, out);

    // 生成响应
    if (4 + out.size() > k_max_msg) {
        out.clear();
        out_err(out, ERR_2BIG, "response too big");
    }
    uint32_t wlen = (uint32_t)out.size();
    memcpy(&conn->wbuf[0], &wlen, 4);
    memcpy(&conn->wbuf[4], out.data(), out.size());
    conn->wbuf_size = 4 + wlen;

    // 移除已处理的请求
    size_t remain = conn->rbuf_size - 4 - len;
    if (remain) {
        memmove(conn->rbuf, &conn->rbuf[4 + len], remain);
    }
    conn->rbuf_size = remain;

    // 改变状态
    conn->state = STATE_RES;
    state_res(conn);

    // 继续外层循环
    return (conn->state == STATE_REQ);
}

/**
 * @brief 尝试填充缓冲区
 */
static bool try_fill_buffer(Conn *conn) {
    // 尝试填充缓冲区
    assert(conn->rbuf_size < sizeof(conn->rbuf));
    ssize_t rv = 0;
    do {
        size_t cap = sizeof(conn->rbuf) - conn->rbuf_size;
        rv = read(conn->fd, &conn->rbuf[conn->rbuf_size], cap);
    } while (rv < 0 && errno == EINTR);
    if (rv < 0 && errno == EAGAIN) {
        // 得到EAGAIN，停止
        return false;
    }
    if (rv < 0) {
        msg("read() error");
        conn->state = STATE_END;
        return false;
    }
    if (rv == 0) {
        if (conn->rbuf_size > 0) {
            printf("客户端 %u: 关闭了连接（有未处理数据）\n", conn->conn_id);
        } else {
            printf("客户端 %u: 关闭了连接\n", conn->conn_id);
        }
        conn->state = STATE_END;
        return false;
    }

    conn->rbuf_size += (size_t)rv;
    assert(conn->rbuf_size <= sizeof(conn->rbuf));

    // 尝试处理请求
    while (try_one_request(conn)) {}
    return (conn->state == STATE_REQ);
}

/**
 * @brief 状态机：请求状态
 */
static void state_req(Conn *conn) {
    while (try_fill_buffer(conn)) {}
}

/**
 * @brief 尝试刷新缓冲区
 */
static bool try_flush_buffer(Conn *conn) {
    ssize_t rv = 0;
    do {
        size_t remain = conn->wbuf_size - conn->wbuf_sent;
        rv = write(conn->fd, &conn->wbuf[conn->wbuf_sent], remain);
    } while (rv < 0 && errno == EINTR);
    if (rv < 0 && errno == EAGAIN) {
        // 得到EAGAIN，停止
        return false;
    }
    if (rv < 0) {
        msg("write() error");
        conn->state = STATE_END;
        return false;
    }
    conn->wbuf_sent += (size_t)rv;
    assert(conn->wbuf_sent <= conn->wbuf_size);
    if (conn->wbuf_sent == conn->wbuf_size) {
        // 响应完全发送，回到请求状态
        conn->state = STATE_REQ;
        conn->wbuf_sent = 0;
        conn->wbuf_size = 0;
        return false;
    }
    // 还有更多数据要发送
    return true;
}

/**
 * @brief 状态机：响应状态
 */
static void state_res(Conn *conn) {
    while (try_flush_buffer(conn)) {}
}

/**
 * @brief 连接IO处理
 */
static void connection_io(Conn *conn) {
    // 更新空闲计时器
    conn->idle_start = get_monotonic_usec();
    dlist_detach(&conn->idle_list);
    dlist_insert_before(&g_data.idle_list, &conn->idle_list);

    if (conn->state == STATE_REQ) {
        state_req(conn);
    } else if (conn->state == STATE_RES) {
        state_res(conn);
    } else {
        assert(0);  // 不应该到达这里
    }
}

/**
 * @brief 接受新连接
 */
static void accept_new_conn(int fd) {
    // 接受
    struct sockaddr_in client_addr = {};
    socklen_t socklen = sizeof(client_addr);
    int connfd = accept(fd, (struct sockaddr *)&client_addr, &socklen);
    if (connfd < 0) {
        msg("accept() error");
        return;  // 错误
    }

    // 设置新连接fd为非阻塞模式
    fd_set_nb(connfd);
    // 创建Conn结构
    struct Conn *conn = (struct Conn *)malloc(sizeof(struct Conn));
    if (!conn) {
        close(connfd);
        return;
    }
    conn->fd = connfd;
    conn->state = STATE_REQ;
    conn->conn_id = g_data.next_conn_id++;  // 分配连接ID
    conn->rbuf_size = 0;
    conn->wbuf_size = 0;
    conn->wbuf_sent = 0;
    conn->idle_start = get_monotonic_usec();
    dlist_insert_before(&g_data.idle_list, &conn->idle_list);
    
    printf("客户端 %u: 建立了连接\n", conn->conn_id);
}

/**
 * @brief 处理空闲连接
 */
static void process_idle_connections() {
    const uint64_t k_idle_timeout_us = 60 * 1000000;  // 60秒
    uint64_t now_us = get_monotonic_usec();
    
    while (!dlist_empty(&g_data.idle_list)) {
        Conn *next = container_of(g_data.idle_list.next, Conn, idle_list);
        uint64_t next_us = next->idle_start + k_idle_timeout_us;
        if (next_us >= now_us + 1000) {  // 1ms的容差
            break;
        }
        
        printf("客户端 %u: 超时关闭连接\n", next->conn_id);
        close(next->fd);
        dlist_detach(&next->idle_list);
        free(next);
    }
}

/**
 * @brief 主函数
 */
int main() {
    // 初始化全局数据
    hm_init(&g_data.db);  // 初始化哈希表
    dlist_init(&g_data.idle_list);
    thread_pool_init(&g_data.tp, 4);
    g_data.next_conn_id = 1;  // 初始化连接计数器

    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) {
        die("socket()");
    }

    int val = 1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val));

    // 绑定
    struct sockaddr_in addr = {};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(1234);
    addr.sin_addr.s_addr = ntohl(0);    // 通配符地址 0.0.0.0
    int rv = bind(fd, (const sockaddr *)&addr, sizeof(addr));
    if (rv) {
        die("bind()");
    }

    // 监听
    rv = listen(fd, SOMAXCONN);
    if (rv) {
        die("listen()");
    }

    // 设置监听fd为非阻塞模式
    fd_set_nb(fd);

    // 事件循环
    std::vector<struct pollfd> poll_args;
    while (true) {
        // 准备poll()的参数
        poll_args.clear();
        // 监听fd总是第一个
        struct pollfd pfd = {fd, POLLIN, 0};
        poll_args.push_back(pfd);

        // 连接fds
        DList *cur = g_data.idle_list.next;
        while (cur != &g_data.idle_list) {
            Conn *conn = container_of(cur, Conn, idle_list);
            cur = cur->next;
            
            struct pollfd pfd = {};
            pfd.fd = conn->fd;
            pfd.events = (conn->state == STATE_REQ) ? POLLIN : POLLOUT;
            pfd.events = pfd.events | POLLERR;
            poll_args.push_back(pfd);
        }

        // poll等待活动
        int timeout_ms = 1000;
        int rv = poll(poll_args.data(), (nfds_t)poll_args.size(), timeout_ms);
        if (rv < 0) {
            die("poll");
        }

        // 处理活动连接
        for (size_t i = 1; i < poll_args.size(); ++i) {
            if (poll_args[i].revents) {
                Conn *conn = container_of(g_data.idle_list.next, Conn, idle_list);
                // 找到对应的连接
                DList *cur = g_data.idle_list.next;
                for (size_t j = 1; j < i; ++j) {
                    cur = cur->next;
                }
                conn = container_of(cur, Conn, idle_list);
                connection_io(conn);
                if (conn->state == STATE_END) {
                    // 客户端关闭，或出现错误
                    close(conn->fd);
                    dlist_detach(&conn->idle_list);
                    free(conn);
                }
            }
        }

        // 尝试接受新连接
        if (poll_args[0].revents) {
            accept_new_conn(fd);
        }

        // 处理计时器
        process_timers();
        process_idle_connections();
    }

    return 0;
} 